package com.atguigu.spzx.cart.listener;

import com.alibaba.fastjson.JSON;
import com.atguigu.spzx.cart.service.CartService;
import com.atguigu.spzx.common.handler.interceptor.SpzxServiceAuthInterceptor;
import com.atguigu.spzx.common.util.consts.SpzxConst;
import com.atguigu.spzx.model.entity.h5.CartInfo;
import com.atguigu.spzx.model.entity.h5.UserInfo;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.data.redis.core.BoundHashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.util.Map;
import java.util.concurrent.TimeUnit;

@Component
@Slf4j
public class CartKafkaListener {
    @Resource
    StringRedisTemplate stringRedisTemplate;
    @Resource
    RedisTemplate redisTemplate;
    @Resource
    CartService cartService;
    /*
        监听订单创建成功的kafka消息
            - 删除购物车中创建订单涉及的选中的购物项
                购物车数据使用userId拼接的唯一的key   skuId映射购物项 存储
     */
    @KafkaListener(topics = SpzxConst.SPZX_ORDER_TOPIC , //要消费的主题
//            topicPartitions = @TopicPartition(),
            groupId = "cart_order_topic_cg01") //当前消费者属于哪个分组
    public void spzxOrderTopicListener1(ConsumerRecord<String,String> record ,
                                        Acknowledgment ack){
        String value = record.value();
        Map<String,Object> map = JSON.parseObject(value, Map.class);
        if("1".equals(map.get("type").toString())){
            //获取到订单创建的消息
            String orderId = map.get("orderId").toString();
            long userId = Long.parseLong(map.get("userId").toString());
            BigDecimal totalAmount = new BigDecimal(map.get("totalAmount").toString());
            log.info("获取到订单创建的消息：{},{},{}" , orderId,userId,totalAmount);
            //orderId 已经存到redis中了，代表当前消费者已经消费过该条记录
//            if(stringRedisTemplate.hasKey("spzx:cart:kafka:token:"+orderId)){
//                return;
//            }
            //orderId的消息第一次接收到
//            stringRedisTemplate.
            //setnx(k,v): set not exists  设置kv当key不存在时
            Boolean absent = stringRedisTemplate.opsForValue().setIfAbsent("spzx:cart:kafka:token:" + orderId, "1", 30, TimeUnit.MINUTES);
            if(!absent){//设置key失败的请求直接return  orderId 已经存到redis中了，代表当前消费者已经消费过该条记录
                return;
            }
            //处理第一次获取到订单创建成功的消息的业务
            //获取用户购物车
            BoundHashOperations<String,String, CartInfo> ops = redisTemplate.boundHashOps("spzx:cart:" + userId);
            ops.entries().forEach((skuIdStr , cartInfo)->{
                if(cartInfo.getIsChecked().intValue() == 1){
                    ops.delete(skuIdStr);
                }
            });
        }


        //手动提交offset
        ack.acknowledge();
    }

    @KafkaListener(topics = SpzxConst.SPZX_ORDER_TOPIC , //要消费的主题
            groupId = "cart_order_topic_cg01") //当前消费者属于哪个分组
    public void spzxOrderTopicListener2(ConsumerRecord<String,String> record ,
                                        Acknowledgment ack){
        String value = record.value();
        Map<String,Object> map = JSON.parseObject(value, Map.class);
        if("1".equals(map.get("type").toString())){
            //获取到订单创建的消息
            String orderId = map.get("orderId").toString();
            long userId = Long.parseLong(map.get("userId").toString());
            BigDecimal totalAmount = new BigDecimal(map.get("totalAmount").toString());
            log.info("获取到订单创建的消息：{},{},{}" , orderId,userId,totalAmount);
            //orderId 已经存到redis中了，代表当前消费者已经消费过该条记录
//            if(stringRedisTemplate.hasKey("spzx:cart:kafka:token:"+orderId)){
//                return;
//            }
            //orderId的消息第一次接收到
//            stringRedisTemplate.
            //setnx(k,v): set not exists  设置kv当key不存在时
            Boolean absent = stringRedisTemplate.opsForValue().setIfAbsent("spzx:cart:kafka:token:" + orderId, "1", 30, TimeUnit.MINUTES);
            if(!absent){//设置key失败的请求直接return  orderId 已经存到redis中了，代表当前消费者已经消费过该条记录
                return;
            }
            //处理第一次获取到订单创建成功的消息的业务
            //获取用户购物车
            BoundHashOperations<String,String, CartInfo> ops = redisTemplate.boundHashOps("spzx:cart:" + userId);
            ops.entries().forEach((skuIdStr , cartInfo)->{
                if(cartInfo.getIsChecked().intValue() == 1){
                    ops.delete(skuIdStr);
                }
            });
        }


        //手动提交offset
        ack.acknowledge();
    }

    @KafkaListener(topics = SpzxConst.SPZX_ORDER_TOPIC , //要消费的主题
            groupId = "cart_order_topic_cg01") //当前消费者属于哪个分组
    public void spzxOrderTopicListener3(ConsumerRecord<String,String> record ,
                                        Acknowledgment ack){
        String value = record.value();
        Map<String,Object> map = JSON.parseObject(value, Map.class);
        if("1".equals(map.get("type").toString())){
            //获取到订单创建的消息
            String orderId = map.get("orderId").toString();
            long userId = Long.parseLong(map.get("userId").toString());
            BigDecimal totalAmount = new BigDecimal(map.get("totalAmount").toString());
            log.info("获取到订单创建的消息：{},{},{}" , orderId,userId,totalAmount);
            //orderId 已经存到redis中了，代表当前消费者已经消费过该条记录
//            if(stringRedisTemplate.hasKey("spzx:cart:kafka:token:"+orderId)){
//                return;
//            }
            //orderId的消息第一次接收到
//            stringRedisTemplate.
            //setnx(k,v): set not exists  设置kv当key不存在时
            Boolean absent = stringRedisTemplate.opsForValue().setIfAbsent("spzx:cart:kafka:token:" + orderId, "1", 30, TimeUnit.MINUTES);
            if(!absent){//设置key失败的请求直接return  orderId 已经存到redis中了，代表当前消费者已经消费过该条记录
                return;
            }
            //处理第一次获取到订单创建成功的消息的业务
            //获取用户购物车
            BoundHashOperations<String,String, CartInfo> ops = redisTemplate.boundHashOps("spzx:cart:" + userId);
            ops.entries().forEach((skuIdStr , cartInfo)->{
                if(cartInfo.getIsChecked().intValue() == 1){
                    ops.delete(skuIdStr);
                }
            });
        }


        //手动提交offset
        ack.acknowledge();
    }
}
